package com.zili.tableapiandsql;

import com.zili.source.ClickSource;
import com.zili.source.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author : ranzlupup
 * @since : 2023/3/15 20:39
 */
public class SimpleTableExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.读取数据，得到DataStream
        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        }));

        // 2.创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 3.将DataStream转换成Table
        Table table = tableEnv.fromDataStream(stream);

        tableEnv.createTemporaryView("table1", table);
        // 4.直接写SQL进行转换
        // Table resTable = tableEnv.sqlQuery("select * from " + table);
        Table resTable = tableEnv.sqlQuery("select * from table1");

        // TableResult resTable = tableEnv.executeSql("select * from " + table);
        // resTable.print();
        tableEnv.toDataStream(resTable).print();

        env.execute();
    }
}
